/*
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or (at
 * your option) any later version.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301, USA.
 * */

// AmzServer - Tototoy's server model.
// Author: tonyjobmails@gmail.com (tonyhack).
//
// Define class SimpleRingQueue.
//
// SimpleRingQueue implement a queue uses to communicate between threads,
// The data transfered by thread is stored being a pointer in this ring queue.
// This SimpleRingQueue is also a ""double-thread safe" implement just like
// RingQueueSingleModelBoundless, the different thing is this SimpleRingQueue
// just saves a object pointer, and RingQueueSingleModelBoundless saves the
// data(or a object).
//

#ifndef __SIMPLE__RING__QUEUE__H
#define __SIMPLE__RING__QUEUE__H

#include <queue>
#include <deque>

#include <ext/pool_allocator.h>

#include "common.h"
#include "logging.h"

namespace amz {

template <typename Type, uint32 kStaticQueueSize = 1024>
class SimpleRingQueue {
  typedef std::pair<bool, Type*> ValuePair;
  typedef __gnu_cxx::__pool_alloc<ValuePair> PoolAllocator;
  typedef std::queue<ValuePair, std::deque<ValuePair, PoolAllocator> > DynamicQueue;

 public:
  SimpleRingQueue();
  ~SimpleRingQueue();

  // Use only for producer thread.
  void Add(Type *element);
  // Use only for consumer thread.
  Type *Remove();

  // Flush to ring queue.
  // Use only for producer thread.
  inline bool Flush();

  // Clear the element.
  void Clear();

 private:
  // Shift element into ring queue.
  // Use only for producer thread.
  bool Shift2Ring();

  void RemoveOut();

  // Dynamic queue.
  DynamicQueue dynamic_queue_;
  // Ring queue.
  ValuePair ring_[kStaticQueueSize];
  // Write/Read pointer.
  volatile uint32 write_pointer_;
  volatile uint32 read_pointer_;
};

template <typename Type, uint32 kStaticQueueSize>
SimpleRingQueue::SimpleRingQueue() : write_pointer_(0), read_pointer_(0) {
  // Initialize.
  uint32 pos = 0;
  for(; pos < kStaticQueueSize; ++pos) {
    this->ring_[pos].first = false;
  }
}

template <typename Type, uint32 kStaticQueueSize>
SimpleRingQueue::~SimpleRingQueue() {}

template <typename Type, uint32 kStaticQueueSize>
void SimpleRingQueue::Add(Type *element) {
  if(this->Shift2Ring() && this->ring_[this->write_pointer_].first == false) {
    this->ring_[this->write_pointer_].second = element;
    this->ring_[this->write_pointer_].first = true;
    this->write_pointer_ = ++this->write_pointer_ % kStaticQueueSize;
  } else {
    this->dynamic_queue_.push(element);
  }
}

template <typename Type, uint32 kStaticQueueSize>
Type *SimpleRingQueue::Remove() {
  Type *element = NULL;
  if(this->ring_[this->read_pointer_].first == true) {
    element = this->ring_[this->read_pointer_].second;
    this->ring_[this->read_pointer_].first = false;
    this->read_pointer_ = ++this->read_pointer_ % kStaticQueueSize;
  }
  return element;
}

template <typename Type, uint32 kStaticQueueSize>
bool SimpleRingQueue::Flush() {
  return this->Shift2Ring();
}

template <typename Type, uint32 kStaticQueueSize>
bool SimpleRingQueue::Shift2Ring() {
  while(this->dynamic_queue_.empty() == false) {
    if(this->ring_[this->write_pointer_].first == false) {
      // Have space to write.
      this->ring_[this->write_pointer_].second = element;
      this->ring_[this->write_pointer_].first = true;
      this->write_pointer_ = ++this->write_pointer_ % kStaticQueueSize;
      // Shift one, pop it.
      this->dynamic_queue_.pop();
    } else {
      return false;
    }
  }
  return true;
}

template <typename Type, uint32 kStaticQueueSize>
bool SimpleRingQueue::Clear() {
  while(this->Shift2Ring() == false) {
    this->RemoveOut();
  }
  this->RemoveOut();
}

template <typename Type, uint32 kStaticQueueSize>
void SimpleRingQueue::RemoveOut {
  if(this->ring_[this->read_pointer_].first == true) {
    this->ring_[this->read_pointer_].first = false;
    this->read_pointer_ = ++this->read_pointer_ % kStaticQueueSize;
  }
}

}  // namespace amz

#endif  // __SIMPLE__RING__QUEUE__H

